
/**
 * @ProjectName: 
 * @Copyright: 2016 lisheng All Right Reserved.
 * toughheart@163.com
 * 2016年9月19日 下午9:40:54
 * @Description: 本内容未经本人允许禁止使用、转发.
 */
 
package com.ls.fw.data.hbase.service.impl;

import java.io.Closeable;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.hbase.MasterNotRunningException;
import org.apache.hadoop.hbase.ZooKeeperConnectionException;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;

import com.ls.fw.data.hbase.bean.HbaseStatus;
import com.ls.fw.data.hbase.bean.Record;
import com.ls.fw.data.hbase.client.HBaseClient;
import com.ls.fw.data.hbase.exception.HbaseException;
import com.ls.fw.data.hbase.helper.HbaseHelper;
import com.ls.fw.data.hbase.helper.HbaseHelper.ColumnHandler;
import com.ls.fw.data.hbase.helper.HbaseHelper.ConvertHandler;
import com.ls.fw.data.hbase.service.HBaseService;
import com.ls.fw.data.hbase.support.HBaseTable;
import com.ls.fw.data.hbase.support.HBaseTable.RowHandler;
import com.ls.fw.data.hbase.support.TableName;


/**
 * 
 * @author lisheng
 * 2016年9月19日 下午9:40:54
 * @version V1.0 
 */
public class HBaseServiceImpl implements HBaseService  , Closeable,InitializingBean{
	protected final Logger logger = LoggerFactory.getLogger(this.getClass());
	
	private HBaseClient client;
	private int clientPort = 2181;
	private String quorum;
	private boolean checkTable = false;//false 表示表一定存在，如果真的不存在就自动创建
	
	@Override
	public List<Map<String,Object>> select(String table
			,String rowkeyName, ColumnHandler columnHandler, String... ids) throws Exception{
		if(!checkTable){
			create(table);
		}
		final List<Get> list = new ArrayList<Get>();
		if(ids != null && ids.length>0){
			for (String id : ids) {
				list.add(new Get(Bytes.toBytes(id)));
			}
		}
		return get(table, columnHandler, list,rowkeyName);
	}

	private List<Map<String, Object>> get(String table, ColumnHandler columnHandler
			, final List<Get> list,String rowkeyName)
			throws Exception {
		if(!checkTable){
			create(table);
		}
		HBaseTable t = client.table(TableName.valueOf(table));
		Result[] result = t.get(list);
		List<Map<String,Object>> validData = null;
		if(result!=null && result.length>0){
			validData = new ArrayList<Map<String,Object>>(result.length);
			for (Result result2 : result) {
				Record record = HbaseHelper.parse(result2, HBaseService.FAMILY, columnHandler);
				Map<String,Object> data = record.get(HBaseService.FAMILY);
				if(StringUtils.isNotBlank(rowkeyName)){
					data.put(rowkeyName, record.getId());
				}
				validData.add(data);
			}
		}
		t.close();
		return validData;
	}

	@Override
	public List<Map<String,Object>> select(String table, ColumnHandler columnHandler
			, List<String> ids, String rowkeyName) throws Exception{
		final List<Get> list = new ArrayList<Get>();
		if(ids != null && ids.size()>0){
			for (String id : ids) {
				list.add(new Get(Bytes.toBytes(id)));
//				list.add(new Get(Bytes.toBytes(RowUtils.get(table,id))));
			}
		}
		return get(table, columnHandler, list, rowkeyName);
	}

	@Override
	public List<String> add(String table,List<Record> map,ConvertHandler handler) throws Exception{
		List<String> idlist = new ArrayList<String>();
		List<Put> list = new ArrayList<Put>();
		if(map != null && !map.isEmpty()){
			for (Record record : map) {
				if(StringUtils.isBlank(record.getId())){
					throw new HbaseException("主键不能为空~！");
				}
//				record.setId(RowUtils.get(table, record.getId()));
				idlist.add(record.getId());
				list.add(HbaseHelper.parse(record, null));
			}
		}
		if(list != null && !list.isEmpty()){
			if(!checkTable){
				create(table);
			}
			HBaseTable t = client.table(TableName.valueOf(table));
			t.put(list);
			t.close();
		}
		return idlist;
	}

	@Override
	public String add(String table,Record record, ConvertHandler handler) throws Exception{
		if(StringUtils.isBlank(record.getId())){
			throw new HbaseException("主键不能为空~！");
		}
		if(!checkTable){
			create(table);
		}
//		record.setId(RowUtils.get(table, record.getId()));
		HBaseTable t = client.table(TableName.valueOf(table));
		t.put(HbaseHelper.parse(record, handler));
		t.close();
		return record.getId();
	}
	@Override
	public String update(String table,Record record, ConvertHandler handler) throws Exception{
		if(StringUtils.isBlank(record.getId())){
			throw new HbaseException("主键不能为空~！");
		}
		record.setId(record.getId());
		if(!checkTable){
			create(table);
		}
		HBaseTable t = client.table(TableName.valueOf(table));
		t.put(HbaseHelper.parse(record, handler));
		t.close();
		return record.getId();
	}

	@Override
	public void scan(String table,String start,String end, RowHandler rowHandler){
		Scan scan = new Scan();
		if(StringUtils.isNotBlank(start)){
			scan.setStartRow(Bytes.toBytes(start));
		}
		if(StringUtils.isNotBlank(end)){
			scan.setStopRow(Bytes.toBytes(end));
		}
		scan.setCacheBlocks(false);
		if(!checkTable){
			create(table);
		}
		HBaseTable t = client.table(TableName.valueOf(table));
		t.scan(scan , rowHandler);
		t.close();
	}

	@Override
	public List<HbaseStatus> getMasterStatus( ) throws MasterNotRunningException, ZooKeeperConnectionException, IOException{
		return client.admin().getMasterStatus();
	}
	@Override
	public List<HbaseStatus> getRegionServerStatus( ) throws IOException{
		return client.admin().getRegionServerStatus();
	}
	@Override
	public void empty(String table){
		client.empty(TableName.valueOf(table));
	}
	
	@Override
	public void drop(String table){
		client.delete(TableName.valueOf(table));
	}
	

	@Override
	public void create(String table){
		client.create(TableName.valueOf(table), HBaseService.FAMILY);
	}
	
	@Override
	public void delete(String table,String... ids){
//		String[] rowkeys = null;
//		if(ids != null && ids.length>0){
//			rowkeys = new String[ids.length];
//			int i = 0;
//			for (String id : ids) {
//				rowkeys[i++] = RowUtils.get(table,id);
//			}
//		}
		if(!checkTable){
			create(table);
		}
		HBaseTable t = client.table(TableName.valueOf(table));
		t.delete(ids);
		t.close();
	}

	@Override
	public void close(){
		client.close();
	}
	
	
	public int getClientPort() {
		return this.clientPort;
	}
	public void setClientPort(int clientPort) {
		this.clientPort = clientPort;
	}
	public String getQuorum() {
		return this.quorum;
	}
	public void setQuorum(String quorum) {
		this.quorum = quorum;
	}
	@Override
	public void afterPropertiesSet() throws Exception {
		if(client == null)
			client = (HBaseClient.Builder.create().set(HBaseClient.Parameter.ZK_QUORUM, this.quorum )
				.setInt(HBaseClient.Parameter.ZK_PROPERTY_CLIENT_PORT, this.clientPort ).build());
	}
	public HBaseClient getClient() {
		return client;
	}
	public void setClient(HBaseClient client) {
		this.client = client;
	}

	@Override
	public boolean exists(String table, String id) {
		Get get = new Get(Bytes.toBytes(id));
		if(!checkTable){
			create(table);
		}
		HBaseTable t = client.table(TableName.valueOf(table));
		boolean f = t.exists(get );
		t.close();
		return f;
	}

	@Override
	public String getMaxRowKey(String table) throws IOException, Exception {
		TableName tableName = TableName.valueOf(table);
		Scan scan = new Scan();
		scan.setMaxResultSize(1);
		scan.setBatch(1);
		scan.setReversed(true);
		if(!checkTable){
			create(table);
		}
		ResultScanner r = client.table(tableName).scan(scan );
		Record r1 = HbaseHelper.parse(r.next(), null);
		return r1.getId();
	}

	@Override
	public String getMinRowKey(String table) throws IOException, Exception {
		TableName tableName = TableName.valueOf(table);
		Scan scan = new Scan();
		scan.setMaxResultSize(1);
		scan.setBatch(1);
		scan.setReversed(false);
		if(!checkTable){
			create(table);
		}
		ResultScanner r = client.table(tableName).scan(scan );
		Record r1 = HbaseHelper.parse(r.next(), null);
		return r1.getId();
	}

	public boolean isCheckTable() {
		return checkTable;
	}

	public void setCheckTable(boolean checkTable) {
		this.checkTable = checkTable;
	}
	
}
